package io.github.junxworks.junx.stream.handler;

import io.github.junxworks.junx.stream.constants.DefaultParams;
import io.github.junxworks.junx.stream.constants.DefaultTopics;
import io.github.junxworks.junx.event.EventChannel;
import io.github.junxworks.junx.event.EventContext;

import java.util.Collection;
import java.util.Map;

public abstract class AbstractSingleDataDataHandler extends AbstractDataHandler {
    /**
     * @param event   当前事件上下文
     * @param channel 事件通道
     * @throws Exception
     */
    @Override
    public void handleEvent(EventContext event, EventChannel channel) throws Exception {
        incrCount();
        try {
            boolean shouldPass = shouldPassEvent();
            Object data = event.getData(DefaultParams.PARAM_DATA);
            if (data instanceof Collection) {
                for (Object obj : ((Collection) data)) {
                    try {
                        Object res = handle(obj, globalParams);
                        if (shouldPass && res != null) {
                            event.setData(DefaultParams.PARAM_DATA, res);
                            channel.publish(outputTopic, event);
                        }
                    } catch (Exception e) {
                        event.setData(DefaultParams.PARAM_EXCEPTION, e);
                        event.setData(DefaultParams.PARAM_GLOBAL_PARAMS, globalParams);
                        channel.publish(DefaultTopics.TOPIC_EXCEPTION, event);
                    }
                }
            } else {
                Object res = handle(data, globalParams);
                if (shouldPass && res != null) {
                    event.setData(DefaultParams.PARAM_DATA, res);
                    channel.publish(outputTopic, event);
                }
            }

        } catch (Exception e) {
            event.setData(DefaultParams.PARAM_EXCEPTION, e);
            event.setData(DefaultParams.PARAM_GLOBAL_PARAMS, globalParams);
            channel.publish(DefaultTopics.TOPIC_EXCEPTION, event);
        }finally {
            decrCount();
        }
    }

    public abstract Object handle(Object data, Map<String, Object> globalParams);
}
